package com.zhu.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.zhu.bean.TradeTrademarkCategoryUserRefundBean;
import com.zhu.config.KafkaTopicConfig;
import com.zhu.function.DIMAsyncFunction;
import com.zhu.function.DIMJoinFunction;
import com.zhu.utils.ClickHouseUtil;
import com.zhu.utils.DataFormatUtil;
import com.zhu.utils.ZhuKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import javax.xml.crypto.Data;
import java.time.Duration;
import java.util.Collections;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Locale;
import java.util.concurrent.TimeUnit;

/**
 * 交易域 用户 品类 品牌粒度 退单各窗口汇总表
 * 五个粒度  123 品类  品牌 用户id sku id
 */
public class DWSTradeTrademarkCategoryUserRefundWindowApp {

    public static void main(String[] args) throws Exception {

        // env
        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
        executionEnvironment.setParallelism(4);  //kafka topic_partition

        //kafka_source dwd_order_refund
        String topic = KafkaTopicConfig.KAFKA_DWD_ORDER_REFUND_TOPIC;
        String groupId = "dws_trade_trademark_category_user_refund" + KafkaTopicConfig.KAFKA_GROUP_ID_LAST_NAME;
        DataStreamSource<String> kafkaDStream = executionEnvironment.addSource(ZhuKafkaUtil.getFlinkKafkaConsumer(topic, groupId));

        //json
        SingleOutputStreamOperator<JSONObject> filterJsonObjDStream = kafkaDStream.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String data, Collector<JSONObject> collector) throws Exception {
                try {
                    JSONObject jsonObject = JSON.parseObject(data);
                    collector.collect(jsonObject);
                } catch (Exception exception) {
                    exception.printStackTrace();
                    System.out.println("Dirty>>>" + data);
                }
            }
        });

        //todo 去重  orderRefund 然后转换为java bean
        //没有左链接 产生的重复数据  但是会有 相同orderid  粒度 为  品牌 用户 品类  最后合并 set 集合
        SingleOutputStreamOperator<TradeTrademarkCategoryUserRefundBean> mapToBeanAndFilterOrderDStream = filterJsonObjDStream.map(
                jsonObject -> {
                    String orderId = jsonObject.getString("order_id");
                    String userId = jsonObject.getString("user_id");
                    String skuId = jsonObject.getString("sku_id");
                    Long ts = jsonObject.getLong("row_op_ts");
                    return TradeTrademarkCategoryUserRefundBean.builder()
                            .orderIdSet(new HashSet<String>(
                                    Collections.singleton(orderId)
                            ))
                            .userId(userId)
                            .ts(ts)
                            .skuId(skuId)
                            .build();
                }
        );

        // 补充与分组相关的字段 异步IO 多线程 查询  redis优化
        SingleOutputStreamOperator<TradeTrademarkCategoryUserRefundBean> withSkuInfoDStream = AsyncDataStream.unorderedWait(mapToBeanAndFilterOrderDStream, new DIMAsyncFunction<TradeTrademarkCategoryUserRefundBean>("DIM_SKU_INFO") {
                    @Override
                    public void join(TradeTrademarkCategoryUserRefundBean input, JSONObject dimInfo) {
                        //查到 的hbase json 数据
                        input.setCategory3Id(dimInfo.getString("category3_Id"));
                        input.setTradeMarkId(dimInfo.getString("tm_id"));

                    }

                    @Override
                    public String getKey(TradeTrademarkCategoryUserRefundBean input) {
                        return input.getSkuId();
                    }
                }
                , 100,
                TimeUnit.SECONDS);

        //tradeMark
        SingleOutputStreamOperator<TradeTrademarkCategoryUserRefundBean> withWaterMarkDStream = withSkuInfoDStream.assignTimestampsAndWatermarks(WatermarkStrategy.<TradeTrademarkCategoryUserRefundBean>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                .withTimestampAssigner(new SerializableTimestampAssigner<TradeTrademarkCategoryUserRefundBean>() {
                    @Override
                    public long extractTimestamp(TradeTrademarkCategoryUserRefundBean tradeTrademarkCategoryUserRefundBean, long l) {
                        return tradeTrademarkCategoryUserRefundBean.getTs();
                    }
                }));

        //分组

        KeyedStream<TradeTrademarkCategoryUserRefundBean, String>  keyedByUserSkuInfoDStream = withSkuInfoDStream.keyBy(new KeySelector<TradeTrademarkCategoryUserRefundBean, String>() {
            @Override
            public String getKey(TradeTrademarkCategoryUserRefundBean tradeTrademarkCategoryUserRefundBean) throws Exception {
                return tradeTrademarkCategoryUserRefundBean.getTradeMarkId() +
                        tradeTrademarkCategoryUserRefundBean.getCategory3Id() +
                        tradeTrademarkCategoryUserRefundBean.getUserId();

            }
        });

        //聚合
        WindowedStream<TradeTrademarkCategoryUserRefundBean, String, TimeWindow> windowkeyedByUserAndSkuInfoDStream = keyedByUserSkuInfoDStream.window(TumblingEventTimeWindows.of(Time.seconds(10)));
        SingleOutputStreamOperator<TradeTrademarkCategoryUserRefundBean> reduceDStream = windowkeyedByUserAndSkuInfoDStream.reduce(
                new ReduceFunction<TradeTrademarkCategoryUserRefundBean>() {
                    @Override
                    public TradeTrademarkCategoryUserRefundBean reduce(TradeTrademarkCategoryUserRefundBean value1, TradeTrademarkCategoryUserRefundBean value2) throws Exception {
                        value1.getOrderIdSet().addAll(value2.getOrderIdSet());
                        return value1;
                    }
                },
                new ProcessWindowFunction<TradeTrademarkCategoryUserRefundBean, TradeTrademarkCategoryUserRefundBean, String, TimeWindow>() {
                    @Override
                    public void process(String s, ProcessWindowFunction<TradeTrademarkCategoryUserRefundBean, TradeTrademarkCategoryUserRefundBean, String, TimeWindow>.Context context, Iterable<TradeTrademarkCategoryUserRefundBean> iterable, Collector<TradeTrademarkCategoryUserRefundBean> collector) throws Exception {
                        String start = DataFormatUtil.toYmdHms(context.window().getStart());
                        String end = DataFormatUtil.toYmdHms(context.window().getEnd());
                        TradeTrademarkCategoryUserRefundBean next = iterable.iterator().next();
                        next.setStt(start);
                        next.setEdt(end);
                        next.setRefundCount((long) next.getOrderIdSet().size());
                        next.setTs(System.currentTimeMillis());
                        collector.collect(next);
                    }
                }
        );

        //关联无关维度
        SingleOutputStreamOperator<TradeTrademarkCategoryUserRefundBean> joinWithTradeMarkNDStream = AsyncDataStream.unorderedWait(reduceDStream,
                new DIMAsyncFunction<TradeTrademarkCategoryUserRefundBean>("DIM_BASE_TRADEMARK") {
                    @Override
                    public void join(TradeTrademarkCategoryUserRefundBean input, JSONObject dimInfo) {
                        input.setTradeMarkName(dimInfo.getString("tm_name"));
                    }

                    @Override
                    public String getKey(TradeTrademarkCategoryUserRefundBean input) {
                        return input.getTradeMarkId();
                    }
                },
                100, TimeUnit.SECONDS);

        SingleOutputStreamOperator<TradeTrademarkCategoryUserRefundBean> joinWithBaseC3DStream = AsyncDataStream.unorderedWait(joinWithTradeMarkNDStream, new DIMAsyncFunction<TradeTrademarkCategoryUserRefundBean>("DIM_BASE_CATEGORY3") {
                    @Override
                    public void join(TradeTrademarkCategoryUserRefundBean input, JSONObject dimInfo) {
                        input.setCategory3Name(dimInfo.getString("name"));
                        input.setCategory2Id(dimInfo.getString("category2_id").toUpperCase(Locale.ROOT));
                    }

                    @Override
                    public String getKey(TradeTrademarkCategoryUserRefundBean input) {
                        return input.getCategory3Id();
                    }
                },
                100, TimeUnit.SECONDS);

        SingleOutputStreamOperator<TradeTrademarkCategoryUserRefundBean> joinWithBaseC2DStream = AsyncDataStream.unorderedWait(joinWithBaseC3DStream, new DIMAsyncFunction<TradeTrademarkCategoryUserRefundBean>("DIM_BASE_CATEGORY2") {
            @Override
            public void join(TradeTrademarkCategoryUserRefundBean input, JSONObject dimInfo) {
                input.setCategory2Name(dimInfo.getString("name"));
                input.setCategory1Id(dimInfo.getString("category1_id").toUpperCase(Locale.ROOT));
            }

            @Override
            public String getKey(TradeTrademarkCategoryUserRefundBean input) {
                return input.getCategory2Id();
            }
        }, 100, TimeUnit.SECONDS);

        SingleOutputStreamOperator<TradeTrademarkCategoryUserRefundBean> resultDStream = AsyncDataStream.unorderedWait(joinWithBaseC2DStream, new DIMAsyncFunction<TradeTrademarkCategoryUserRefundBean>("DIM_BASE_CATEGORY1") {
            @Override
            public void join(TradeTrademarkCategoryUserRefundBean input, JSONObject dimInfo) {
                input.setCategory1Name(dimInfo.getString("name"));
            }

            @Override
            public String getKey(TradeTrademarkCategoryUserRefundBean input) {
                return input.getCategory1Id();
            }
        }, 100, TimeUnit.SECONDS);


        resultDStream.print(">>>>");
        resultDStream.addSink(ClickHouseUtil.getClickHouseSinkFunction(
                "inset into dws_trade_trademark_category_user_refund_window values((?,?,?,?,?,?,?,?,?,?,?,?,?)"
        ));
        //job
        executionEnvironment.execute("DWSTrademarkCategoryUserRefundWindowApp");
    }



}
